package com.scala.Class

import java.{lang, util}

import breeze.numerics.pow
import com.java.Enum.PopularityAnalysis
import com.java.bean.TestInfoBean
import com.java.config.RedisConfig
import com.java.util.{ErrorRangeUtil, PatternUtil, SerializeUtil, TestRedisPool}
import com.scala.CaseClass.RentalBuildingInfo
import org.apache.spark.sql.{SQLContext, SparkSession}
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

import scala.collection.mutable



/*
* 后期优化 https://www.cnblogs.com/cord/p/9530404.html
* https://www.cnblogs.com/duanxz/p/4520627.html
* https://github.com/ZhuXS/Spring-Shiro-Spark/blob/master/Server/src/main/java/com/zhuxs/result/config/ApplicationConfig.java
*  redis  设置密码
*做JedisPool时遇到的问题还真不少，遇到了入下问题，这里提供一个解决办法，不一定使用所有人遇到的情况，因为可能有很多人使用的是在虚拟机里配置的redis，我用的配置是centos服务器上的redis，问题的原因可能就是出自这里，因为如果是本地的话，redis.conf里面的配置默认是 bind 127.0.0.1，而远程连接的话要将这句配置注释掉，并且非本地连接时会默认进入保护模式？需要设置密码或者修改其他的配置以达到访问的目的
* */
/**
 * @Author: ZhangJin
 * @Date: 2020/9/8 11:21
 */
@Service(value = "cal")
class cal @Autowired()(val session: SparkSession) extends Serializable {

  // TODO:   获取数据
  def getSource(LocalName: String, codelaterrentinfodata: String): util.Map[Array[Byte], Array[Byte]] = {
    // TODO:   加载hive数据进map集合  ------ LocalName 地方名
    val hc: SQLContext = session.sqlContext
    import session.implicits._
    // TODO:   调用java函数判断redis中是否存在  hset(String key,String item,Object value)
    import scala.collection.JavaConverters._
    val Jedis = TestRedisPool.getJedisPoolInstance(RedisConfig.redisIp, Integer.parseInt(RedisConfig.redisPort),RedisConfig.redisPassword).getResource
    if (Jedis.exists(LocalName.getBytes)) {
      val bytesToBytes: util.Map[Array[Byte], Array[Byte]] = Jedis.hgetAll(LocalName.getBytes)
      bytesToBytes
    } else {
      try {
        // TODO: 没有就去hive中取
        hc.sql("select * from copewithdata." + codelaterrentinfodata).as[RentalBuildingInfo].rdd.foreachPartition(
          eachPartition => {
            val value = TestRedisPool.getJedisPoolInstance(RedisConfig.redisIp, Integer.parseInt(RedisConfig.redisPort),RedisConfig.redisPassword).getResource
            eachPartition.foreach(each => {
              value.hset(LocalName.getBytes(), each.UUid.getBytes(), mutable.Map("UUid" -> each.UUid, "LocationDetail" -> each.LocationDetail, "PriceByMonth" -> each.PriceByMonth, "Introduction" -> each.Introduction, "Title" -> each.Title, "WebSite" -> each.WebSite, "Type" -> each.Type, "StateNow" -> each.StateNow, "Sheer" -> each.Sheer, "LocationDetail" -> each.LocationDetail, "AreaOfHouse" -> each.AreaOfHouse, "Width" -> each.Width, "Depth" -> each.Depth, "LocationOfLatAndLong" -> each.LocationOfLatAndLong, "Local" -> each.Local, "NumOfSeePepple" -> PopularityAnalysis.getPopularity(each.NumOfSeePepple).getIndex).asJava.toString().getBytes())
            }
            )
            // TODO:   释放资源
            TestRedisPool.returnResource(value)
          }
        )
        val bytesToBytes: util.Map[Array[Byte], Array[Byte]] = Jedis.hgetAll(LocalName.getBytes())
        // TODO:   释放资源
        TestRedisPool.returnResource(Jedis)
        bytesToBytes
      } catch {
        case ex: Exception => {
          println("No find table ...")
          null
        }
      }
    }

  }

  // TODO: 第一步筛选计算
  def calChangeStepFirst(testInfoBean: TestInfoBean,Data:util.Map[Array[Byte], Array[Byte]], nowTimeAsNameForZset: java.lang.Long): Unit ={
   if (Data == null){
    println("无数据")
   }
    import scala.collection.JavaConverters._
    /*  map的value变为list   想把每个value的key变为每个list中的其中一员 */
    val list = new util.ArrayList[Array[Byte]](Data.values())
    val user1 = new util.HashMap[String,Double]
    val map1 = SerializeUtil.convertToMap(testInfoBean).asScala
    user1.put("Type",map1("type").toDouble)
    user1.put("StateNow",map1("stateNow").toDouble)
    user1.put("Title",map1("title").toDouble)
    session.sparkContext.parallelize(list.asScala,10)
      .foreachPartition(
      {
        eachPartition=>{
          val Jedis = TestRedisPool.getJedisPoolInstance(RedisConfig.redisIp, Integer.parseInt(RedisConfig.redisPort),RedisConfig.redisPassword).getResource
          val ZsetMap = new util.HashMap[String, java.lang.Double]()
          eachPartition.foreach(each=>{
            try {
              var stringToString: util.Map[String, String] = SerializeUtil.mapStringToMap(new String(each,"UTF-8"))
              // TODO:  第一步筛选
              if (isNotFirstChoose(testInfoBean,stringToString)){
                // TODO:   第二步筛选  直接入 redis --> zset
                println("UUid为: " + stringToString.get("UUid")+"的数据,通过第一步筛选。进行第二次筛选")
                var calResult: Double = calChangeStepSecond(user1.asScala, stringToString.asScala)
                ZsetMap.put(stringToString.get("UUid"),calResult)
              }
            } catch {
              case ex: Exception => {
                ex.printStackTrace()
                System.err.println("出现错误")
              }
            }
          }
          )
          /*  每个分区存入Zset中  */
          if (ZsetMap.size()!=0)
          Jedis.zadd(nowTimeAsNameForZset.toString,ZsetMap)
          /*  设置失活时间  */
          Jedis.expire(nowTimeAsNameForZset.toString,1000)
          TestRedisPool.returnResource(Jedis)
      }
      }
    )
  }

  // TODO:   第一步筛选的条件限制  -->  进深 宽度 面积 价位   --> 硬性要求
  def isNotFirstChoose(testInfoBean: TestInfoBean,stringToString: util.Map[String, String]):Boolean = {
    ErrorRangeUtil.isNotLocateRangeForDepth(testInfoBean.getDepth,stringToString.get("Depth"))&&
         ErrorRangeUtil.isNotLocateRangeForWidth(testInfoBean.getWidth,stringToString.get("Width"))&&
         ErrorRangeUtil.isNotLocateRangeForArea(testInfoBean.getAreaOfHouse,stringToString.get("AreaOfHouse"))&&
         ErrorRangeUtil.isNotLocateRangeForPrice(testInfoBean.getPriceByMonth,stringToString.get("PriceByMonth"))
  }

/* ----  yyq ----  */
  // TODO: 第二步筛选计算  计算结果放入redis中Zset
  def calChangeStepSecond(user1: scala.collection.mutable.Map[String, Double], map2:scala.collection.mutable.Map[String, String]): Double = {
    val user2 = new util.HashMap[String,Double]
    user2.put("Type",map2("Type").toDouble)
    user2.put("StateNow",map2("StateNow").toDouble)
    user2.put("Title",map2("Title").toDouble)
    //获得第一个用户的评分
    import scala.collection.JavaConverters._
    val user1Source: Vector[Double] = user1.values.toVector
    //获得第二个用户的评分
    val user2Source: Vector[Double] = user2.asScala.values.toVector
    PearsonCorrelationSimilarity(user1Source,user2Source)
  }



  def PearsonCorrelationSimilarity(vec1 : Vector[Double], vec2 : Vector[Double]): Double = {
    val sum_vec1 = vec1.sum
    val sum_vec2 = vec2.sum

    val square_sum_vec1 = vec1.map(x => x * x).sum
    val square_sum_vec2 = vec2.map(x => x * x).sum

    val zipVec = vec1.zip(vec2)

    val product = zipVec.map(x => x._1 * x._2).sum
    val numerator = product - (sum_vec1 * sum_vec2 / vec1.length)

    val dominator = pow((square_sum_vec1 - pow(sum_vec1, 2) / vec1.length) * (square_sum_vec2 - pow(sum_vec2, 2) / vec2.length), 0.5)

    if(dominator == 0)
      0
    else
      numerator / (dominator * 1.0)
  }
}
